package com.aotain.metis.bolt;

import java.sql.Connection;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.apollo.IPScanUpdate;
import com.aotain.common.CommonDB;
import com.aotain.common.DbPoolMultiple;
import com.aotain.storm.AbstractBolt;
import com.mysql.jdbc.Statement;

/**
 * 威胁评估统计
 * @ClassName: ThreatenEvaluateBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2017年3月16日 下午4:57:43 
 *
 */
public class ThreatenEvaluateBolt extends AbstractBolt{

	private static final long serialVersionUID = 1L;


	String sip = null;
	String abnormaltype = null;
	Map<String,Integer> evalMap = new HashMap<String,Integer>();//评分Map
	String dbJson;


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(isTickTuple(tuple)) {
				emitCountingData(collector);  	
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println(" MonAbnReverseStatisBolt exception"+e.getMessage());
			Logger.getRootLogger().error("MonAbnReverseStatisBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}

	private void emitCountingData(OutputCollector collector) {

		java.sql.Statement stmt = null;
		Connection conn = DbPoolMultiple.getInstance().getConnJson(dbJson);
		String insert = "insert into METIS_THREATEN_EVALUATE (REPORTTIME,IP,EVALUATE) values ('%s','%s','%s')";
		SimpleDateFormat sdfh_ = new SimpleDateFormat("yyyy-MM-dd HH:00:00");
		Date date = new Date();
		String dateStrH_ = sdfh_.format(date);
		String sip = "";
		Integer eval = 0;
		
		try {
			
			stmt = conn.createStatement();
			conn.setAutoCommit(false);
			
			for(Entry<String,Integer> entry : evalMap.entrySet()) {
				
				sip = entry.getKey();
				eval = entry.getValue();
				String sql = String.format(insert, new Object[]{dateStrH_,sip,eval});
				stmt.execute(sql);
				
			}
			conn.commit();
			
		} catch (SQLException e) {
			Logger.getLogger(getClass()).error("insert ip list error",e);
		} finally {
			CommonDB.closeDBConnection(conn, stmt, null);
		}
		
		
		evalMap.clear();
	}

	private void countInLocal(Tuple tuple) {

		if(tuple != null ) {  
			sip = tuple.getStringByField("sip");
			abnormaltype = tuple.getStringByField("abntype");
			
			int eval = 90;
			switch (abnormaltype) {
			case "CC":
				eval = 59;
				break;
			case "BOTS":
				eval = 59;
				break;
			case "SCANATTACK":
				eval = 59;
				break;
			case "ABNSESSION":
				eval = 79;
				break;
			case "XSS":
				eval = 90;
				break;
			case "SQLINJECTION":
				eval = 90;
				break;
			default:
				break;
			}
			/**
			 * 一个小时内存分数最小的情况
			 */
			if(!"".equals(evalMap.get(sip)) && evalMap.get(sip)!= null) {
				int tmpEval = evalMap.get(sip);
				if(tmpEval<eval) {
					eval = tmpEval;
				}
			}
			evalMap.put(sip, eval);
		}
	}

	@Override
	public void cleanup() {
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60*60);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub

		dbJson = stormConf.get("dbJsonMetis").toString();
	}

}
